package com.shujia.flink.source

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._

object Demo8RichSource {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(4)

    val ds: DataStream[String] = env.addSource(new TestRichSource)


    ds.print()

    env.execute()

  }
}

class TestRichSource extends RichSourceFunction[String] {

  var conn: Connection = _

  /**
    * open  方法只会执行异常  在tun之前执行
    *
    * 大莲姐网络连接
    */
  override def open(parameters: Configuration): Unit = {
    Class.forName("com.mysql.jdbc.Driver")
    conn = DriverManager.getConnection("jdbc:mysql://master:3306/test", "root", "123456")

  }

  /**
    * 回收资源  也只执行异常  在run之后执行
    *
    */
  override def close(): Unit = {
    conn.close()
  }

  override def run(ctx: SourceFunction.SourceContext[String]): Unit = {
    val stat: PreparedStatement = conn.prepareStatement("select * from student")
    val resultSet: ResultSet = stat.executeQuery()

    while (resultSet.next()) {
      val id: String = resultSet.getString("id")
      val name: String = resultSet.getString("name")
      val age: Int = resultSet.getInt("age")
      val gender: String = resultSet.getString("gender")
      val clazz: String = resultSet.getString("clazz")


      val line: String = s"$id,$name,$age,$gender,$clazz"

      //将数据发送到下游
      ctx.collect(line)
    }
  }

  override def cancel(): Unit = {

  }
}
